package com.feidee.fdhadoop.hbase;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.constant.CommonConstant;
import com.feidee.fdcommon.util.TimeUtils;
import com.feidee.fdhadoop.configUpdater.ConfigUpdater;
import com.feidee.fdhadoop.constant.Constant;
import com.feidee.fdhadoop.logCollect.LogCollectService;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.*;


public class HBaseUtil {
    protected static final Logger logger = LoggerFactory.getLogger(HBaseUtil.class);
    private static Connection conn;

    private static String APPLICATION_NAME = "";
    private static String CLINET_IP = "";

    // 初始化连接
    static {
        ConfigUpdater.getInstance("hbaseHA", new HBaseConfigUpdaterCallBack())
                .executeCallBack();
        CLINET_IP = getIp();
        APPLICATION_NAME = CustomConfiguration.getString("application.name", "");
    }

    /**
     * 获取指定行和列的数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @return
     */
    public static String get(String tableName, String rowKey, String family, String qualifier) {
        long start = System.currentTimeMillis();
        Table t = null;
        try {
            t = getTable(tableName);
            return HBaseUtilHelper.get(rowKey, family, qualifier, t);
        } catch (Exception e) {
            logger.error("HBaseUtil执行get异常！", e);
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, rowKey, family, qualifier);
        }
        return null;
    }


    /**
     * 获取指定行的数据
     *
     * @param tableName
     * @param rowKey
     * @return
     */
    public static Map<String, Map<String, String>> get(String tableName, String rowKey) {
        long start = System.currentTimeMillis();
        Map<String, Map<String, String>> results = null;
        Table t = null;
        try {
            t = getTable(tableName);
            results = HBaseUtilHelper.get(rowKey, t);
        } catch (Exception e) {
            logger.error("HBaseUtil执行get异常！", e);
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, rowKey, "", "");
        }
        return results;
    }


    /**
     * 获取指定行和列族的数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @return
     */
    public static Map<String, String> get(String tableName, String rowKey, String family) {
        long start = System.currentTimeMillis();
        Map<String, String> result = null;
        Table t = null;
        try {
            t = getTable(tableName);
            result = HBaseUtilHelper.get(rowKey, family, t);
        } catch (Exception e) {
            logger.error("HBaseUtil执行get异常！", e);
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, rowKey, family, "");
        }
        return result;
    }


    /**
     * 获取指定行和部分列的数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifierSet
     * @return
     */
    public static Map<String, String> get(String tableName, String rowKey, String family, Set<String> qualifierSet) throws Exception {
        long start = System.currentTimeMillis();
        Map<String, String> result = null;
        Table t = null;
        try {
            t = getTable(tableName);
            result = HBaseUtilHelper.get(rowKey, family, qualifierSet, t);
        } catch (Exception e) {
            logger.error("HBaseUtil执行get异常！", e);
            throw e;
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, rowKey, family, qualifierSet.toString());
        }
        return result;
    }

    /**
     * 其它api不能满足时使用，比如需要使用filter的情况
     *
     * @param tableName
     * @param get
     * @return
     */
    public static Result get(String tableName, Get get) {
        long start = System.currentTimeMillis();
        Result result = null;
        Table t = null;
        try {
            t = getTable(tableName);
            result = t.get(get);
        } catch (Exception e) {
            logger.error("HBaseUtil执行get异常！", e);
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, Bytes.toString(get.getRow()), "", "");
        }
        return result;
    }

    /**
     * 其它api不能满足时使用，比如需要使用filter的情况
     *
     * @param tableName
     * @param gets
     * @return
     */
    public static Result[] get(String tableName, List<Get> gets) {
        long start = System.currentTimeMillis();
        Result[] results = null;
        Table t = null;
        try {
            t = getTable(tableName);
            results = t.get(gets);
        } catch (Exception e) {
            logger.error("HBaseUtil执行get异常！", e);
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, Bytes.toString(gets.get(0).getRow()), "", "", Constant.HBASE_OP_TYPE_GET, gets.size());
        }
        return results;
    }

    /**
     * 写入单列数据
     *
     * @param tableName
     * @param rowkey
     * @param family
     * @param qualifier
     * @param value
     * @throws Exception
     */
    public static void put(String tableName, String rowkey, String family, String qualifier, String value) throws Exception {
        long start = System.currentTimeMillis();
        Table t = null;
        try {
            t = getTable(tableName);
            Put put = new Put(Bytes.toBytes(rowkey));
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
            t.put(put);
        } catch (Exception e) {
            logger.error("HBaseUtil执行put异常！", e);
            throw e;
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, rowkey, family, qualifier, Constant.HBASE_OP_TYPE_PUT, 1);
        }
    }

    /**
     * 写入单行数据
     *
     * @param tableName
     * @param rowkey
     * @param family
     * @param valueMap
     * @throws Exception
     */
    public static void put(String tableName, String rowkey, String family, Map<String, String> valueMap) throws Exception {
        long start = System.currentTimeMillis();
        Table t = null;
        try {
            t = getTable(tableName);
            HBaseUtilHelper.put(rowkey, family, valueMap, t);
        } catch (Exception e) {
            logger.error("HBaseUtil执行put异常！", e);
            throw e;
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, rowkey, family, "", Constant.HBASE_OP_TYPE_PUT, 1);
        }
    }


    /**
     * 写入多行数据
     *
     * @param tableName
     * @param family
     * @param recordList
     * @throws Exception
     */
    public static void put(String tableName, String family, List<Map<String, String>> recordList) throws Exception {
        long start = System.currentTimeMillis();
        Table t = null;
        try {
            t = getTable(tableName);
            HBaseUtilHelper.put(family, recordList, t);
        } catch (Exception e) {
            logger.error("HBaseUtil执行put异常！", e);
            throw e;
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, "", "", "", Constant.HBASE_OP_TYPE_PUT, recordList.size());
        }
    }


    /**
     * 写入数据，用于不需要value的场景
     *
     * @param tableName
     * @param rowkey
     * @throws Exception
     */
    public static void put(String tableName, String rowkey) throws Exception {
        put(tableName, rowkey, HBaseUtilHelper.DEFAULT_FAMILY);
    }

    /**
     * 写入数据，用于不需要value的场景
     *
     * @param tableName
     * @param rowkey
     * @param family
     * @throws Exception
     */
    public static void put(String tableName, String rowkey, String family) throws Exception {
        Map<String, String> values = new HashMap<>();
        values.put(HBaseUtilHelper.DEFAULT_QUALIFIER, "");
        put(tableName, rowkey, family, values);
    }

    /**
     * 删除指定行的列
     *
     * @param tableName
     * @param rowkey
     * @param family
     * @param qualifier
     * @return
     */
    public static boolean del(String tableName, String rowkey, String family, String qualifier) {
        long start = System.currentTimeMillis();
        Table t = null;
        try {
            t = getTable(tableName);
            Delete del = new Delete(Bytes.toBytes(rowkey));

            if (qualifier != null) {
                del.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            } else if (family != null) {
                del.addFamily(Bytes.toBytes(family));
            }
            t.delete(del);
            return true;
        } catch (Exception e) {
            logger.error("HBaseUtil执行del异常！", e);
        } finally {
            closeTable(t);
            sendRequestLog(start, tableName, rowkey, family, qualifier, Constant.HBASE_OP_TYPE_DEL, 1);
        }
        return false;
    }

    /**
     * 删除指定行
     *
     * @param tableName
     * @param rowKey
     * @return
     */
    public static boolean del(String tableName, String rowKey) {
        return del(tableName, rowKey, null, null);
    }

    /**
     * 删除指定行的列族
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @return
     */
    public static boolean del(String tableName, String rowKey, String family) {
        return del(tableName, rowKey, family, null);
    }

    /**
     * 现有api不满足需求时使用；
     * 由于可能会在线切换集群，所以不要缓存Table；
     *
     * @param table
     * @return
     * @throws Exception
     */
    public static Table getTable(String table) throws Exception {
        try {
            if (conn != null) {
                return conn.getTable(TableName.valueOf(table));
            }
        } catch (Exception e) {
            logger.error("HBaseUtil get table异常！", e);
            throw e;
        }
        return null;
    }

    public static void closeTable(Table table) {
        if (table != null) {
            try {
                table.close();
            } catch (Exception e) {
                logger.error("HBaseUtil close table异常！", e);
            }
        }
    }

    /**
     * 进行Admin相关操作时使用；
     * 由于可能会在线切换集群，所以不要缓存Connection；
     */
    public static Connection getConn() {
        return conn;
    }

    public static void setConn(Connection newConn) {
        close();
        conn = newConn;
    }

    public static void close() {
        if (conn != null) {
            try {
                conn.close();
            } catch (Exception e) {
                logger.error("HBaseUtil 关闭连接异常！", e);
            }
        }
    }

    public static void sendRequestLog(long start, String tableName, String rowkey, String cf
            , String qualifiers) {
        sendRequestLog(start, tableName, rowkey, cf, qualifiers, Constant.HBASE_OP_TYPE_GET, 1);
    }

    public static void sendRequestLog(long start, String tableName, String rowkey, String cf
            , String qualifiers, String opType, int opSize) {
        long costTime = System.currentTimeMillis() - start;
        try {
            TableName table = TableName.valueOf(tableName);
            HbaseLogModel hbaseLogModel = new HbaseLogModel();
            hbaseLogModel.setClient_ip(CLINET_IP);
            hbaseLogModel.setTable_name(tableName);
            hbaseLogModel.setRowkey(rowkey);
            hbaseLogModel.setColumn_family(cf);
            hbaseLogModel.setQualifiers(qualifiers);
            if (!"".equals(rowkey)) {
                HRegionLocation hrl = conn.getRegionLocator(table).getRegionLocation(Bytes.toBytes(rowkey));
                hbaseLogModel.setRegion_server(hrl.getHostname());
                hbaseLogModel.setRegion_name(hrl.getRegion().getEncodedName());
            }
            hbaseLogModel.setCost_time(Long.toString(costTime));
            hbaseLogModel.setRequest_time(TimeUtils.getCurrentTime());
            hbaseLogModel.setApplication_name(APPLICATION_NAME);
            hbaseLogModel.setOp_type(opType);
            hbaseLogModel.setOp_size(Integer.toString(opSize));
            LogCollectService.sendLog(Constant.LOG_APT, Constant.LOG_BUS_HBASEUTIL, hbaseLogModel);
        } catch (Exception e) {
            logger.error("hbaseutil 日志上报异常~", e);
        }
    }

    private static String getIp() {
        String ip = "";
        try {
            Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
            Collection<InetAddress> addresses = new ArrayList<InetAddress>();

            while (networkInterfaces.hasMoreElements()) {
                NetworkInterface networkInterface = networkInterfaces.nextElement();
                Enumeration<InetAddress> inetAddresses = networkInterface.getInetAddresses();
                while (inetAddresses.hasMoreElements()) {
                    InetAddress inetAddress = inetAddresses.nextElement();
                    addresses.add(inetAddress);
                }
            }

            for (InetAddress address : addresses) {
                if (!address.isLoopbackAddress() && !address.getHostAddress().contains(":"))
                    ip = address.getHostAddress();
            }
        } catch (Exception e) {
            logger.error("获取本机ip异常~", e);
        }
        logger.info("ip:{}", ip);
        return ip;
    }

    public static String getApplicationName() {
        return APPLICATION_NAME;
    }

    public static void setApplicationName(String applicationName) {
        APPLICATION_NAME = applicationName;
    }

}
